/*
--------------------------------------------------------------------------------
				Ak-Lib. 
			 Project       	: Ak-Lib
			 Group         	: Ak-Lib R&D Team
			 Author        	: Ocean.Wang
			 Create Date 	: 05-01-2010
--------------------------------------------------------------------------------
Code Name 	:  A.K 	~v~
Description	:  Yet Another Cross-Platform Lib.
Cautions    	:
--------------------------------------------------------------------------------
Change History:
Version		: 0.1
Building 		: 05/01/10
10-05.01 first draft, build, debug and test.
--------------------------------------------------------------------------------
*/

#include "ak_mc.h"

/* ------------------------------------------------------------------
**	Mc Bucket Server Context:
**	1. Read. Dispatch to Task.
**	Get.
**	2. Write.CbEntry in the Transport will Direct Access the Hash Tbl to reduce the Memcpy. 
**	Set. /Del.
*/
#define _MC_SVR_JOBQ_RING 	0X1000
#define _MC_SVR_JOBQ_CNT 	0X2000
#define _MC_SVR_JOBQ_BAT 	0X100

typedef struct _t_mc_sJobNode{
	_t_sysJobNode 		jobNode;

	int  				code;		/* Operation & Status Code */
	_t_mc_trans 		trans;
	
	/* key & PiggyBack */
	unsigned long long 	key;
	unsigned long long 	piggyback;
}_t_mc_sJobNode;

typedef struct _t_mc_sJobQ{
	/* Hash Group: Kaaa.....!, Here We only need to Lock the Process Que */
	_t_sysSemB 		opWake;

	/* Job Buffers: Hi-Pri & Md-Pri */
	_t_sysJobQ 		jobQ;
	
	/* Static Info */
	int 			flag;
	_t_sysTask 		thread;
	_t_sysTaskAttr 	tAttr;

	struct _t_mc_sNode * pScNode;

}_t_mc_sJobQ;


#define _MC_SVR_THREAD  		8
#define _MC_SVR_TRANS 			5

#define _MC_SVR_TASK_STACK 		0X100000
#define _MC_SVR_TASK_PRI 		90

typedef struct _t_mc_sNode{
	_t_bnode 			node;

	_t_hTblLru 			hTblCtx;

	/* Transport */
	_t_mc_trans 		trans[_MC_SVR_TRANS];

	/* Task */
	_t_mc_sJobQ 		task[_MC_SVR_THREAD];

	
	/* Statistics */
	unsigned long 		stat_Proc;	
	unsigned long 		stat_Job;	

	unsigned long 		stat_Hit;
	unsigned long 		stat_Miss;

	unsigned long 		stat_OOM;
	unsigned long 		stat_DispF;
}_t_mc_sNode;

#define   MC_SC_KEYGEN(pKey)		_SHTbl_KeyGen(pKey, (sizeof(unsigned long long)/sizeof(unsigned char)), _rHTBL_BUCKET_MAX)
	

/* ------------------------------------------------------------------
**	Mc Server Ctrl Job.
*/
static	int	_McSc_JobProc(_t_mc_sNode * pScNode, long lArg, _t_mc_sJobNode * pJobNode)
{
	int 				ret, index;
	void 			* 	pKey;
	_t_sysLruCt 	* 	pBucket;
	_t_mc_item 		* 	pItem;
	
	_t_sgbuf			sgBuf[2];
	_t_mc_message	*	mcMsg;
	unsigned char 		msgBuf[_MC_MSG_MAXSZ];

	ret 			= -1;
	mcMsg 			= (_t_mc_message *)msgBuf;
	mcMsg->elem.type= _MC_SAP_INTER;
	mcMsg->elem.dir	= _MSGELEM_RSP;

	switch(pJobNode->code){
		case _MC_CMD_GET:
			/* Key */
			mcMsg->data.key 		= pJobNode->key;
			mcMsg->elem.piggyback	= pJobNode->piggyback;
			pKey 					= &(mcMsg->data.key);
			/* Search Item */
			index	= MC_SC_KEYGEN(pKey);
			
			pBucket = &(pScNode->hTblCtx.buckets[index]);
			pItem	= McItem_Get(pBucket, pKey, &(mcMsg->data), _MC_MSG_MAXSZ);
			if(pItem){
				pScNode->stat_Hit += 1;
				
				mcMsg->code 	= _MC_ERR_OK;
				mcMsg->elem.length = sizeof(_t_mc_message) + mcMsg->data.dLen;
	
				sgBuf[0].SGBUF_PTR = (unsigned char *)(mcMsg);
				sgBuf[0].SGBUF_LEN = sizeof(_t_mc_message) + mcMsg->data.dLen;
				McTrans_Tx(&(pJobNode->trans), sgBuf, 1);
				ret = 0;
			}
			else{
				pScNode->stat_Miss += 1;
				
				mcMsg->data.rLen = 0;
				mcMsg->data.dLen = 0;
				mcMsg->code = _MC_ERR_NFOUND;
				mcMsg->elem.length = sizeof(_t_mc_message);

				sgBuf[0].SGBUF_PTR = (unsigned char *)(mcMsg);
				sgBuf[0].SGBUF_LEN = sizeof(_t_mc_message);
				McTrans_Tx(&(pJobNode->trans), sgBuf, 1);
			}
			break;
		default:
			break;
	}

	return ret;
}


#define _MCS_TASK_NAME_FMT 		"McS_%02d"
static 	void * 	_McSc_JobEntry(_t_mc_sJobQ * pTask)
{
	int 			procNum;

	_t_mc_sNode		*	pScNode;
	_t_sysJobQ  	* 	pJobQ;
	_t_sysJobNode   * 	jobNodeA[_MC_SVR_JOBQ_BAT];
	
	if(pTask){
		STask_Attr(&(pTask->tAttr));
		
		pScNode = pTask->pScNode;
		STask_SetName(_MCS_TASK_NAME_FMT, (int)(long)(pTask - pScNode->task));
		
		/* Set the Flag of Task */
		pTask->flag = 1;
		
		while(1){
			SYS_SEM_B_TAKE(&(pTask->opWake), _WAIT_FOREVER);

__LABEL_MCSVR_JOB_REDO:			
			/* 	---------------------------------------------------------------
			**	1. Operations:
			*/
			pJobQ = &(pTask->jobQ);
			pScNode->stat_Proc += 1;
			procNum = SJOBQ_COUNT(pJobQ);
			/* Statistics */
			if(procNum > pScNode->stat_Job)
				pScNode->stat_Job = procNum;
			
			/* Schedule */
			SJobQ_Sche(pJobQ, jobNodeA);
			
			if(!SJOBQ_ISEMPTY(pJobQ)){
				goto __LABEL_MCSVR_JOB_REDO;
			}
			

		}
	}

	return NULL;
}

static 	void 	_McSc_JobInit(_t_mc_sNode * pScNode)
{
	int 			i;
	_t_mc_sJobQ * 	pTask;
	_t_cbentry 	 	jProc;

	_CBENTRY_INIT(&jProc, _McSc_JobProc, pScNode, 0);
	/* Init the Tasks */
	for(i=0, pTask=pScNode->task; i<_MC_SVR_THREAD; i++, pTask++){
		pTask->pScNode = pScNode;
		SYS_SEM_B_INIT(&(pTask->opWake));
		SJobQ_Init(&(pTask->jobQ), 
					&(pTask->opWake),
					&jProc, 
					_MC_SVR_JOBQ_RING,
					_MC_SVR_JOBQ_BAT,
					_MC_SVR_JOBQ_CNT,
					sizeof(_t_mc_sJobNode));
		
		/* Kaaa.....! Now Let's Spawn the Task */
		STASK_CRATE((pTask->thread), 
					_MC_SVR_TASK_STACK,
					_MC_SVR_TASK_PRI,
					(_TASKENTRY)_McSc_JobEntry, 
					(void *)pTask);
	}

	return;
}


int 	McSc_Parse(void * pArg, int lArg, _t_mc_message * pMsg, int msgLen, _t_mc_trans * pTrans)
{
	int 			index;
	size_t 			mLen;
	_t_mc_item 		* pItem;
	_t_sysLruCi		* pClea;
	_t_sysJobQ 		* pJobQ;
	_t_mc_sJobNode	* pJobNode;
	_t_mc_data 		* pData;
	_t_sysLruCt 	* pBucket;
	_t_sysKLMem 	* pKLMem;
	_t_mc_sNode 	* pScNode;

	pScNode = (_t_mc_sNode *)pArg;
	if(pScNode && pMsg){
		pData	= &(pMsg->data);
		switch(pMsg->code){
			case _MC_CMD_GET:
				/* Dispatch */
				index 	= MC_SC_KEYGEN(&(pData->key)); 
				pJobQ= &(pScNode->task[index%_MC_SVR_THREAD].jobQ);

				pJobNode = (_t_mc_sJobNode *)SJobQ_NodeAlloc(pJobQ);
				if(pJobNode){
					pJobNode->trans		= *pTrans;
					pJobNode->piggyback	= pMsg->elem.piggyback;
					pJobNode->key		= pData->key;
					pJobNode->code 		= _MC_CMD_GET;
					
					SJOBQ_ENQUE(pJobQ, &(pJobNode->jobNode), index);
					if(index != 0){
						pScNode->stat_DispF += 1;
						SJobQ_NodeFree(pJobQ, (_t_sysJobNode *)pJobNode);
					}
				}else{
					pScNode->stat_DispF += 1;
				}
				break;
				
			case _MC_CMD_SET:
				index 	= MC_SC_KEYGEN(&(pData->key)); 
				pBucket = &(pScNode->hTblCtx.buckets[index]);
				pKLMem = pBucket->klMem;
				if(pKLMem && pKLMem->mHeap){
					mLen 	= pData->dLen + sizeof(_t_mc_data);
					pItem  	= McItem_Alloc(pKLMem->mHeap, mLen);
					if(pItem){
						pItem->ci.pKey 		= &(pData->key);
						pItem->ci.ti.expire	= pMsg->expire;
						/* Dangerous That We may Free the pItem in the Rep !*/
						pItem = (_t_mc_item *)SLru_Replace(pBucket, &(pItem->ci), &pClea, pData, mLen);
						McItem_KeyBind(pItem);
					}
				}
				else{
					pScNode->stat_OOM += 1;
				}
				break;
				
			case _MC_CMD_DEL:
				index = MC_SC_KEYGEN(&(pData->key)); 
				pBucket = &(pScNode->hTblCtx.buckets[index]);
				McItem_Del(pBucket, &(pData->key));
				break;

			default:
				break;
		}
	}
	
	return 0;
}

void * 		McSc_Init(size_t mLimit, unsigned short port)
{
	_t_sysMHeap  	* pHeap;
	void 		 	* pAlk; 		/* User's Internal Allocator */
	int 			i;
	_t_mc_sNode 	* pScNode;
	_t_sysKLMem 	klMem;
	_t_sysKLFunc 	klFunc;

	pScNode = (_t_mc_sNode *)SYS_MALLOC(sizeof(_t_mc_sNode *));
	if(pScNode && (port>1024)){
		/* Statistic */
		pScNode->stat_Proc 	= 0;
		pScNode->stat_Job 	= 0;
		pScNode->stat_Hit 	= 0;
		pScNode->stat_Miss 	= 0;
		pScNode->stat_OOM 	= 0;
		pScNode->stat_DispF	= 0;

		/* Init the MC Item Heap */
		pAlk = Mc_HeapInit(&(pHeap), mLimit);
		_KLMEM_INIT(&klMem, pAlk, pHeap, McHeap_GC);
		_KLFUNC_INIT(&klFunc, 
			McItem_KeyComp, 
			McItem_DataGet, McItem_DataSet, 
			McItem_Free, McItem_FreeQ);
		
		/* Init the Hash Tbl */
		SHTblLru_Init(&(pScNode->hTblCtx), &klMem, &klFunc);
			
		/* Init the Job Task */
		_McSc_JobInit(pScNode);

		/* Init the Trans Port */
		for(i=0; i<_MC_SVR_TRANS; i++){
			pScNode->trans[i].port = port + i;
		}
	}
	
	return (void *)pScNode;
}

int 		McSc_Dump(void * pArg)
{
	unsigned long 	i, expire;
	_t_mc_sNode 	* pScNode;
	
	pScNode = (_t_mc_sNode *)pArg;
	if(pScNode){
		for(i=0, expire=0; i<_rHTBL_BUCKET_MAX; i++){
			expire += pScNode->hTblCtx.buckets[i].stat_Expire;
			
		}
		printf("McSvr: OOM(%-8lu) Hit(%-8lu) Miss(%-8lu) Expire(%-8lu) DispF(%-8lu)\n", 
			pScNode->stat_OOM,		pScNode->stat_Hit,	  pScNode->stat_Miss,
			expire,
			pScNode->stat_DispF); 
		return 0;
	}
	
	return -1;
}



